package com.yesep.business;

import com.alibaba.dubbo.common.utils.NamedThreadFactory;
import com.yesep.model.MessageBaseModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

/**
 * @program: envm
 * @description: 业务线程
 * @author: quliang
 * @create: 2019-07-30 15:14
 **/
public class BusinessHandlerThread extends Thread {

    private final Logger logger = LoggerFactory.getLogger(BusinessHandlerThread.class);

    static BlockingQueue<MessageBaseModel> businessQueue = new LinkedBlockingQueue<>(10000);


    // 应答报文处理任务 线程池
    final ExecutorService cachedThreadPool = Executors.newFixedThreadPool(
            Runtime.getRuntime().availableProcessors() * 2,
            new NamedThreadFactory("BusinessStreamThreadPool"));

    private long msgCount = 0;

    // 线程是否已经启动
    private boolean isStarted = false;

    public BusinessHandlerThread(String threadName) {
        this.setName(threadName);
    }

    public static void putData(MessageBaseModel data) {
        businessQueue.add(data);
    }


    public void init() {
        try {
            if (!isStarted) {
                this.start();
                isStarted = true;
            }
        } catch (Exception ex) {
            logger.error("BusinessStreamDataThread Error", ex);
        }
    }

    @Override
    public void run() {
        int bitchSize = 0;
        try {
            MessageBaseModel message;
            for (; ; ) {
                if ((message = businessQueue.take()) != null) {
                    cachedThreadPool.execute(new BusinessHandlerTask(message));
                    msgCount++;
                    bitchSize++;
                    if (bitchSize > 100) {
                        Thread.sleep(1L);
                        bitchSize = 0;
                    }
                }
            }
        } catch (Exception ex) {
            logger.error(ex.getMessage(), ex);
        }
    }
}
